#include "MainReactor.h"
#include "SubReactor.h"
#include "TcpConnection.h"
#include <iostream>
#include <unistd.h>
#include <errno.h>
#include <algorithm>
#include <limits>

using std::cout;
using std::endl;
using std::cerr;

MainReactor::MainReactor(const std::string &ip, unsigned short port, int subReactorNum)
: _acceptor(ip, port)
, _epfd(createEpollFd())
, _evtList(1024)
, _isLooping(false)
, _nextSubReactor(0)
{
    for (int i = 0; i < subReactorNum; ++i) {
        _subReactors.emplace_back(std::unique_ptr<SubReactor>(new SubReactor()));
    }
    
    int listenfd = _acceptor.fd();
    addEpollReadFd(listenfd);
    
    int timerfd = _timerQueue.getTimerFd();
    addEpollReadFd(timerfd);
    // setLoadBalanceStrategy(LoadBalanceStrategy::CONSISTENT_HASH);
}

MainReactor::~MainReactor()
{
    close(_epfd);
}

void MainReactor::start()
{
    for (auto &subReactor : _subReactors) {
        subReactor->setNewConnectionCallback(_onNewConnection);
        subReactor->setMessageCallback(_onMessage);
        subReactor->setCloseCallback(_onClose);
        subReactor->start();
    }
    
    _acceptor.ready();
    loop();
}

void MainReactor::stop()
{
    _isLooping = false;
    
    for (auto &subReactor : _subReactors) {
        subReactor->stop();
    }
}

void MainReactor::setAllCallback(TcpConnectionCallback &&cb1,
                                TcpConnectionCallback &&cb2,
                                TcpConnectionCallback &&cb3)
{
    _onNewConnection = std::move(cb1);
    _onMessage = std::move(cb2);
    _onClose = std::move(cb3);
}

void MainReactor::loop()
{
    _isLooping = true;
    while (_isLooping) {
        waitEpollFd();
    }
}

int MainReactor::createEpollFd()
{
    int epfd = epoll_create1(0);
    if (epfd == -1) {
        perror("epoll_create1");
    }
    return epfd;
}

void MainReactor::addEpollReadFd(int fd)
{
    struct epoll_event evt;
    evt.data.fd = fd;
    evt.events = EPOLLIN;
    
    int ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &evt);
    if (-1 == ret) {
        cerr << "addEpollReadFd" << endl;
        return;
    }
}

void MainReactor::waitEpollFd()
{
    int nready = 0;
    do {
        nready = epoll_wait(_epfd, _evtList.data(), _evtList.size(), 2000);
    } while (nready == -1 && errno == EINTR);
    
    if (nready == -1) {
        cerr << "waitEpollFd" << endl;
        return;
    } else if (nready == 0) {
        cout << ">>MainThread_epoll_wait TIMEOUT." << endl;
    } else {
        if (nready == (int)_evtList.size()) {
            _evtList.resize(2 * nready);
        }
        
        for (int i = 0; i < nready; i++) {
            int fd = _evtList[i].data.fd;
            int listenfd = _acceptor.fd();
            int timerfd = _timerQueue.getTimerFd();
            
            if (fd == listenfd) {
                if (_evtList[i].events & EPOLLIN) {
                    handleNewConnection();
                }
            } else if (fd == timerfd) {
                if (_evtList[i].events & EPOLLIN) {
                    handleTimerEvents();
                }
            }
        }
    }
}

void MainReactor::handleNewConnection()
{
    while (true) {
        int connfd = _acceptor.accept();
        if (-1 == connfd) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                break;
            }
            cerr << "handleNewConnection" << endl;
            return;
        }
        
        // 检查连接数限制
        size_t currentConns = getTotalConnections();
        if (currentConns >= _maxConnections) {
            cout << "\n🚨 [连接拒绝] 当前连接数=" << currentConns 
                 << " 达到最大限制=" << _maxConnections 
                 << "，拒绝新连接 fd=" << connfd << endl;
            close(connfd);
            continue;
        }
        
        std::string clientIP = getClientIP(connfd);
        SubReactor* subReactor = getNextSubReactorByStrategy(clientIP);
        
        // 打印分配信息
        const char* strategyName[] = {"轮询", "最少连接", "一致性哈希", "自适应"};
        cout << "[MainReactor] 新连接 fd=" << connfd 
             << " IP=" << clientIP 
             << " 策略=" << strategyName[static_cast<int>(_strategy)]
             << " 连接数=" << subReactor->getConnectionCount() 
             << " 总连接=" << currentConns + 1 << "/" << _maxConnections << endl;
             
        subReactor->addConnection(connfd);
    }
}

void MainReactor::handleTimerEvents()
{
    _timerQueue.handleExpiredTimers();
}

int64_t MainReactor::runAfter(std::chrono::milliseconds delay, TimerCallback &&callback)
{
    return _timerQueue.addTimer(delay, std::chrono::milliseconds(0), std::move(callback));
}

int64_t MainReactor::runEvery(std::chrono::milliseconds interval, TimerCallback &&callback)
{
    return _timerQueue.addTimer(interval, interval, std::move(callback));
}

void MainReactor::cancelTimer(int64_t timerId)
{
    _timerQueue.cancelTimer(timerId);
}

SubReactor* MainReactor::getNextSubReactor()
{
    SubReactor* subReactor = _subReactors[_nextSubReactor].get();
    _nextSubReactor = (_nextSubReactor + 1) % _subReactors.size();
    return subReactor;
}

SubReactor* MainReactor::getMinConnectionReactor()
{
    SubReactor *minLoadReactor = _subReactors[0].get();
    size_t min = minLoadReactor->getConnectionCount();
    
    // 高并发优化：限制遍历频率，避免过度竞争
    static std::atomic<size_t> counter{0};
    size_t current = counter.fetch_add(1);
    
    // 每100次请求才做一次全局最优选择，其他时候使用轮询+简单检查
    if (current % 100 == 0) {
        for (auto &reactor : _subReactors) {
            size_t cnt = reactor->getConnectionCount();
            if (cnt < min) {
                min = cnt;
                minLoadReactor = reactor.get();
            }
        }
    } else {
        // 简化策略：检查相邻的2个reactor，选择较少的那个
        size_t idx = current % _subReactors.size();
        size_t nextIdx = (idx + 1) % _subReactors.size();
        
        size_t cnt1 = _subReactors[idx]->getConnectionCount();
        size_t cnt2 = _subReactors[nextIdx]->getConnectionCount();
        
        minLoadReactor = (cnt1 <= cnt2) ? _subReactors[idx].get() : _subReactors[nextIdx].get();
    }
    
    return minLoadReactor;
}
int64_t getCurrentTimeMs()
{
    auto now = std::chrono::system_clock::now();
    auto duration = now.time_since_epoch();
    return std::chrono::duration_cast<std::chrono::milliseconds>(duration).count();
}
SubReactor *MainReactor::getNextSubReactorMinConns()
{
    auto now = getCurrentTimeMs();

    // 高并发优化：延长缓存时间，减少计算频率
    static constexpr int64_t HIGH_LOAD_CACHE_MS = 200; // 高负载时200ms缓存
    
    // 缓存检测
    if (_cacheMinReactor && (now - _lastUpdateTime) < HIGH_LOAD_CACHE_MS) {
        return _cacheMinReactor;
    }
    
    SubReactor* freeReactor = getMinConnectionReactor(); //获取较空闲reactor
    _cacheMinReactor = freeReactor;
    _lastUpdateTime = now;
    return freeReactor;
}

void MainReactor::setLoadBalanceStrategy(LoadBalanceStrategy strategy) {
    _strategy = strategy;
    
    // 如果切换到一致性哈希，需要初始化哈希环
    if (strategy == LoadBalanceStrategy::CONSISTENT_HASH || 
        strategy == LoadBalanceStrategy::ADAPTIVE) {
        _consistentHash.clear();
        for (size_t i = 0; i < _subReactors.size(); ++i) {
            _consistentHash.addReactor(_subReactors[i].get(), i);
        }
    }
    
    cout << "[MainReactor] 切换负载均衡策略到: " 
         << static_cast<int>(strategy) << endl;
}

SubReactor* MainReactor::getNextSubReactorByStrategy(const std::string& clientIP) {
    switch (_strategy) {
        case LoadBalanceStrategy::ROUND_ROBIN:
            return getNextSubReactor();
            
        case LoadBalanceStrategy::LEAST_CONNECTIONS:
            return getNextSubReactorMinConns();
            
        case LoadBalanceStrategy::CONSISTENT_HASH:
            if (!clientIP.empty()) {
                SubReactor* reactor = _consistentHash.getReactor(clientIP);
                return reactor ? reactor : getNextSubReactorMinConns(); // 回退
            }
            return getNextSubReactorMinConns(); // 回退到最少连接
            
        case LoadBalanceStrategy::ADAPTIVE: {
            // 自适应策略：检查负载差异
            std::pair<size_t, size_t> connRange = getConnectionRange();
            size_t minConn = connRange.first;
            size_t maxConn = connRange.second;
            double loadDiff = (maxConn == 0) ? 0 : (double)(maxConn - minConn) / (minConn + 1);
            
            // 如果负载差异过大（>50%），优先使用最少连接
            if (loadDiff > 0.5) {
                return getNextSubReactorMinConns();
            } else if (!clientIP.empty()) {
                // 负载相对均衡时，使用一致性哈希保持会话亲和性
                SubReactor* reactor = _consistentHash.getReactor(clientIP);
                return reactor ? reactor : getNextSubReactorMinConns();
            }
            return getNextSubReactorMinConns();
        }
        
        default:
            return getNextSubReactorMinConns();
    }
}

std::string MainReactor::getClientIP(int connfd) {
    struct sockaddr_in clientAddr;
    socklen_t len = sizeof(clientAddr);
    
    if (getpeername(connfd, (struct sockaddr*)&clientAddr, &len) == 0) {
        // 对于本地测试(127.0.0.1)，使用端口号增加哈希多样性
        std::string ip = std::string(inet_ntoa(clientAddr.sin_addr));
        if (ip == "127.0.0.1") {
            // 本地测试时，使用 IP:端口 组合增加哈希分散性
            ip += ":" + std::to_string(ntohs(clientAddr.sin_port));
        }
        return ip;
    }
    
    // 获取失败时，使用连接fd作为唯一标识
    return "fallback:" + std::to_string(connfd);
}

std::pair<size_t, size_t> MainReactor::getConnectionRange() {
    if (_subReactors.empty()) {
        return {0, 0};
    }
    
    size_t minConn = _subReactors[0]->getConnectionCount();
    size_t maxConn = minConn;
    
    for (const auto& reactor : _subReactors) {
        size_t count = reactor->getConnectionCount();
        minConn = std::min(minConn, count);
        maxConn = std::max(maxConn, count);
    }
    
    return {minConn, maxConn};
}

size_t MainReactor::getTotalConnections() const {
    size_t total = 0;
    for (const auto& reactor : _subReactors) {
        total += reactor->getConnectionCount();
    }
    return total;
}